/**
    每隔5秒钟，计算过去十分钟手机销量情况
 */
package com.demo.kafka.flink.shop;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.Comparator;
import java.util.Properties;
import java.util.TreeMap;

/*Flink 的不错的实践 https://www.jianshu.com/p/81f0ef7293f7*/
public class TopN {
    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);
        //ProcessingTime是以operator处理的时间为准，它使用的是机器的系统时间来作为data stream的时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "139.9.79.13:19092");
        //从kafka中接收指定的topic的消息作为输入input
        FlinkKafkaConsumer<String> input = new FlinkKafkaConsumer<>("shop-test", new SimpleStringSchema(), properties);

        //设置消费模式为setStartFromEarliest，说明是从最早的记录开始消费
        input.setStartFromEarliest();

        //将input作为数据流stream的source
        DataStream<String> stream = env
                .addSource(input);

        //将输入语句split成一个一个单词并初始化count值为1的Tuple2<String, Integer>类型
        DataStream<Tuple2<String, Integer>> ds = stream
                .flatMap(new LineSplitter());


        DataStream<Tuple2<String, Integer>> wcount = ds
                .keyBy(0)//将同一Key的数据放到同一个分区
                .window(SlidingProcessingTimeWindows.of(Time.seconds(600), Time.seconds(5)))
                //keyBy之后的元素进入一个总时间长度为600s,每5s向后滑动一次的滑动窗口
                .sum(1);// 将相同key的元素的count值相加

        wcount
                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))//(手机1, xx) (手机2,xx)....
                //windowAll定义一个全局窗口，所有key元素进入一个5s长的滚动窗口（选5秒是因为上游的滑动窗口每5s计算一轮数据，topN窗口一次计算只统计一个窗口时间内的变化）
                .process(new TopNAllFunction(5))
                .print();

        env.execute();
    }//

    private static final class LineSplitter implements
            FlatMapFunction<String, Tuple2<String, Integer>> {

        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            //（手机1,1） (手机2，1) （手机3,1）
             out.collect(new Tuple2<String, Integer>(value, 1));
        }

    }

    private static class TopNAllFunction
            extends
            ProcessAllWindowFunction<Tuple2<String, Integer>, String, TimeWindow> {

        private int topSize;

        public TopNAllFunction(int topSize) {

            this.topSize = topSize;
        }

        public void process(

                ProcessAllWindowFunction<Tuple2<String, Integer>, String, TimeWindow>.Context arg0,
                Iterable<Tuple2<String, Integer>> input,
                Collector<String> out) throws Exception {

            TreeMap<Integer, String> treemap = new TreeMap<Integer, String>(
                    new Comparator<Integer>() {

                        @Override
                        public int compare(Integer y, Integer x) {
                            return (x < y) ? -1 : 1;
                        }

                    }); //treemap按照key降序排列，相同count值不覆盖
            int[] pdata=new int[5];
            for (Tuple2<String, Integer> element : input) {
                System.out.println(element);
                if(element.f0.equals("华为")){
                    pdata[0] = element.f1;
                }
                if(element.f0.equals("小米")){
                    pdata[1] = element.f1;
                }
                if(element.f0.equals("OPPO")){
                    pdata[2] = element.f1;
                }
                if(element.f0.equals("Samsung")){
                    pdata[3] = element.f1;
                }
                if(element.f0.equals("iPhone")){
                    pdata[4] = element.f1;
                }

                treemap.put(element.f1, element.f0);
                if (treemap.size() > topSize) { //只保留前面TopN个元素
                    treemap.pollLastEntry();
                }
            }
            //通过柱形图来实现可视化
            BarChart.main((new Timestamp(System.currentTimeMillis())).toString(),pdata[0],pdata[1],pdata[2],pdata[3],pdata[4]);
            //每隔5s输出实时手机销量
            out.collect("=================\n" + new Timestamp(System.currentTimeMillis()) + "\n" + "实时手机销量:\n" + treemap.toString());

        }
    }

}